package com.study;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;

/**
 * @author sjw
 * @description
 * @date 2019-08-19 下午5:21
 */
public class SinkPatchMySQL extends RichSinkFunction<List<User>> {
    private PreparedStatement ps;
    private Connection connection;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.connection = getConnection();
        String sql = "insert into user(age,name) values (?,?)";
        ps = this.connection.prepareStatement(sql);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection != null) connection.close();
        if (ps != null) ps.close();
    }

    @Override
    public void invoke(List<User> users, Context context) throws Exception {
        //组装数据，执行插入操作
        if (users == null) return;
        for (User user : users) {
            ps.setInt(1, user.getId());
            ps.setString(2, user.getName());
            ps.executeUpdate();
        }
    }


    private Connection getConnection() {
        Connection connection = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/demo?user=root&password=123456&useUnicode=true&characterEncoding=UTF-8");

        } catch (ClassNotFoundException | SQLException e) {
            e.printStackTrace();
        }
        return connection;
    }
}
